package oss

import (
	"context"
	"fmt"
	"gitee.com/ymofen/gobase"
	"gitee.com/ymofen/gobase/factory"
	"gitee.com/ymofen/gobase/gotask"
	"gitee.com/ymofen/gobase/gotask/grouptask"
	"gitee.com/ymofen/gobase/subpub"
	"gitee.com/ymofen/sputils"
	"runtime"
	"strings"
	"sync"
	"sync/atomic"
	"time"
)

// 存储数据到OSS中
type OssReader struct {
	startedflag int32
	lk          sync.RWMutex
	caption     string
	buketName   string
	objectKey   string
	workType    string
	confMap     gobase.StrMap
	ex          *subpub.Subchannel

	taskKey string

	closeWg sync.WaitGroup

	op OSSOperator

	readFlag       atomic.Int32
	readWg         sync.WaitGroup
	latestDataTime time.Time
	latestData     []byte
	cacheTtl       time.Duration

	subfn   func(id, channel string, cb subpub.SubFunc)
	unsubfn func(id, channel string) bool

	readTimeOut      time.Duration
	delayDuration    time.Duration
	intervalDuration time.Duration

	statusText string
}

var (
	OssReaderN      atomic.Int32
	OssReaderAliveN atomic.Int32
)

func (this *OssReader) waitForClose() {
	this.lk.Lock()
	this.startedflag = 0
	//conn := this.conn
	//if conn != nil {
	//	go gobase.GoRun(func() {
	//		conn.RequestStop("RequestClose")
	//	})
	//}
	this.lk.Unlock()

	this.closeWg.Wait()
}

func (this *OssReader) Close() error {

	gotask.DefaultPatrolTask().DelTask(this.taskKey)

	OssReaderN.Add(-1)
	this.waitForClose()
	this.lk.Lock()
	defer this.lk.Unlock()
	this.confMap = nil
	this.ex = nil
	this.startedflag = 0
	if this.op != nil {
		gobase.TryClose(this.op)
		this.op = nil
	}
	logger.Debugf("%s closed", this.caption)
	this.subfn = nil
	this.unsubfn = nil
	return nil
}

func (this *OssReader) GetConnString() string {
	return ""
}

func (this *OssReader) HealthyCheck() (healthy gobase.Ternary, msg string) {
	if atomic.LoadInt32(&this.startedflag) == 0 {
		return gobase.TernaryFalse, "shutdown"
	}
	this.lk.RLock()
	defer this.lk.RUnlock()
	if this.op == nil {
		return gobase.TernaryFalse, "oss operator is nil"
	}

	return gobase.TernaryFalse, "shutdown"
}

func (this *OssReader) GetStatus(args ...interface{}) string {
	this.lk.RLock()
	defer this.lk.RUnlock()
	if this.startedflag == 0 {
		return fmt.Sprintf("shutdown")
	}
	return fmt.Sprintf("%s&ex=%s&status=%s", this.confMap.URLEncode0(), this.ex.Status(), this.statusText)
}

func (this *OssReader) subInterval(id, channel string, cb subpub.SubFunc) {
	this.ex.Sub(id, channel, cb)
}

func (this *OssReader) unsubInterval(id, channel string) bool {
	return this.ex.Unsub(id, channel)
}

func (this *OssReader) innerReload() {
	ctx := context.Background()
	if this.readTimeOut > 0 {
		ctx, _ = context.WithTimeout(context.Background(), this.readTimeOut)
	}

	var bb gobase.BytesBuilder
	err := this.op.GetObject(ctx, this.buketName, this.objectKey, &bb)
	if err != nil {
		this.ex.Pub(sputils.SpEvtNotify, 0, sputils.SpTypeError, err)
		return
	}
	this.latestData = bb.Bytes()
	this.latestDataTime = time.Now()
}

func (this *OssReader) checkReloadFromOss() {

	if !this.readFlag.CompareAndSwap(0, 1) {
		this.readWg.Wait()
		return
	}
	defer this.readFlag.Store(0)

	this.readWg.Add(1)
	defer this.readWg.Done()
	//this.closeWg.Add(1)
	//defer this.closeWg.Done()

	// 不需要读取
	if time.Since(this.latestDataTime) < this.cacheTtl {
		return
	}

	this.innerReload()
}

func (this *OssReader) dispForDelay(id, channel string, cb subpub.SubFunc) {
	this.lk.RLock()
	if this.startedflag == 0 {
		this.lk.RUnlock()
		return
	}
	this.closeWg.Add(1)
	defer this.closeWg.Done()
	this.lk.RUnlock()

	this.checkReloadFromOss()
	cb(id, channel, sputils.SpTypeData, this.latestData, "DATA")

}

func (this *OssReader) reloadInterval() {
	this.lk.RLock()
	if this.startedflag == 0 {
		this.lk.RUnlock()
		return
	}
	this.closeWg.Add(1)
	defer this.closeWg.Done()
	this.lk.RUnlock()

	this.innerReload()

	this.ex.Pub(sputils.SpDefaultChannel, 0, sputils.SpTypeData, this.latestData, "DATA")
}

func (this *OssReader) subForOnce(id, channel string, cb subpub.SubFunc) {
	if channel == sputils.SpDefaultChannel {
		if this.delayDuration <= 0 {
			go gobase.GoRun(func() {
				this.dispForDelay(id, channel, cb)
			})
		} else {
			time.AfterFunc(this.delayDuration, func() {
				this.dispForDelay(id, channel, cb)
			})
		}
	}
}

func (this *OssReader) unsubForOnce(id, channel string) bool {
	return true
}

func (this *OssReader) Sub(id, channel string, cb subpub.SubFunc) {
	this.subfn(id, channel, cb)
}

func (this *OssReader) Unsub(id, channel string) bool {
	return this.unsubfn(id, channel)
}

func (this *OssReader) CheckStart() error {
	this.lk.Lock()
	defer this.lk.Unlock()
	if this.startedflag == 1 {
		return nil
	}
	defer func() {
		this.startedflag = 1
	}()

	this.taskKey = fmt.Sprintf("OssReader-%p", this)

	sn := OssReaderN.Add(1)

	this.caption = this.confMap["name"]
	if len(this.caption) == 0 {
		this.caption = fmt.Sprintf("%s-p%d", this.confMap["stream"], sn)
	}

	this.buketName = this.confMap["buketName"]
	this.objectKey = this.confMap["objectKey"]

	implId := this.confMap["impl"]
	if len(implId) == 0 {
		return fmt.Errorf("invalid param, impl is empty")
	}
	implId = fmt.Sprintf("oss.%s", implId)

	ossConf := gobase.NewStrMap()
	ossConf.CopyFromPrefix(this.confMap, "oss.", true)

	obj, err := gobase.CreateFactoryInstance(implId, ossConf)
	if err != nil {
		return err
	}
	this.op = obj.(OSSOperator)

	if this.workType == "once" {
		this.delayDuration = gobase.ParseDurationEx(this.confMap["delay"], time.Second)
		this.cacheTtl = gobase.ParseDurationEx(this.confMap["ttl"], time.Second)
		if this.cacheTtl <= 0 {
			this.cacheTtl = time.Second * 10
		}
	} else {
		interval := gobase.ParseDurationEx(this.confMap["interval"], time.Second*10)
		if interval <= 0 {
			interval = time.Second * 10
		}
		gotask.DefaultPatrolTask().AddTask(this.taskKey, interval, func(id interface{}, args ...interface{}) {
			grouptask.DefaultGroupTask().PostTaskFunc(this.taskKey, this.reloadInterval)
		})

	}

	return nil
}

var (
	DefaultOssReaders = gobase.NewConnectObjectRefStorage(newOssReader)
)

func newOssReader(connstr string) interface{} {
	node := &OssReader{
		confMap: gobase.NewStrMap(),
		ex:      subpub.NewSubchannel(),
	}
	OssReaderAliveN.Add(1)
	runtime.SetFinalizer(node, func(obj interface{}) {
		OssReaderAliveN.Add(-1)
	})
	node.confMap.URLFormDecode(connstr)
	workType := node.confMap["workType"]
	if workType == "once" {
		node.subfn = node.subForOnce
		node.unsubfn = node.unsubForOnce
		node.cacheTtl = time.Second * 10
		node.delayDuration = time.Second
	} else {
		node.subfn = node.subInterval
		node.unsubfn = node.unsubInterval
	}

	return node
}

// conntype=ossreader&oss.impl=s3client
func checkGetOssReader(args ...interface{}) (rval interface{}, err error) {
	if len(args) == 0 {
		err = fmt.Errorf("无效的参数")
		return
	}
	if conf, ok := args[0].(gobase.StrMap); ok {
		s := conf.SortRangeUrlEncode(func(k, v string) bool {
			return k == "buketName" || k == "name" || k == "objectKey" || k == "connrand" || k == "workType" || k == "interval" || k == "ttl" || k == "delay" || k == "impl" || strings.HasPrefix(k, "oss.")
		})
		rval = DefaultOssReaders.CheckGetRef(s)
		return
	} else {
		err = fmt.Errorf("无效的参数")
		return
	}
}

func init() {
	factory.RegisterCreateFunc("sp.ossreader", checkGetOssReader)
}
